package com.example.mymessagequeue.mqserver;

import com.example.mymessagequeue.common.Consumer;
import com.example.mymessagequeue.common.MqException;
import com.example.mymessagequeue.mqserver.core.*;
import com.example.mymessagequeue.mqserver.datacenter.DiskDataCenter;
import com.example.mymessagequeue.mqserver.datacenter.MemoryDataCenter;

import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 通过这个类, 来表示 虚拟主机.
 * 每个虚拟主机下面都管理这自己的 交换机，队列，绑定，消息 数据
 * 同时提供 api 供上层使用
 * 针对 virtualHost 作为业务逻辑的整合者，就需要对于代码中抛出的异常进行处理了。
 */

public class VirtualHost {
    private String virtualHostName;

    public static final String DEFAULT_NAME = "default";

    private MemoryDataCenter memoryDataCenter = new MemoryDataCenter();
    private DiskDataCenter diskDataCenter = new DiskDataCenter();
    private Router router = new Router();
    private ConsumerManager consumerManager = new ConsumerManager(this);

    // 操作交换机的锁对象
    private final Object exchangeLocker = new Object();
    // 操作队列的锁对象
    private final Object queueLocker = new Object();

    public String getVirtualHostName() {
        return virtualHostName;
    }

    public MemoryDataCenter getMemoryDataCenter() {
        return memoryDataCenter;
    }

    public DiskDataCenter getDiskDataCenter() {
        return diskDataCenter;
    }

    public VirtualHost(String name) {
        this.virtualHostName = name;

        // 对于 memoryDataCenter 来说，是不需要额外的初始化操作的。
        // 但是，针对 DiskDataCenter 来说，则需要进行初始化操作。
        diskDataCenter.init();

        // 另外还需要针对硬盘的数据，恢复到内存中
        try {
            memoryDataCenter.recovery(diskDataCenter);
        } catch (IOException | MqException | ClassNotFoundException e) {
            e.printStackTrace();
            System.out.println("[VirtualHost] 恢复内存数据失败！");
        }
    }

    /**
     * 逻辑隔离，获取逻辑隔离后的名字
     * @param name
     * @return
     */
    public String getName(String name) {
        return virtualHostName + "." + name;
    }

    /**
     * 创建交换机, 如果交换机不存在就创建，如果存在直接返回
     *
     * @param exchangeName
     * @param exchangeType
     * @param durable
     * @param autoDelete
     * @param arguments
     * @return 创建成功返回 ture，失败返回 false
     */
    public boolean exchangeDeclare(String exchangeName, ExchangeType exchangeType, boolean durable, boolean autoDelete,
                                   Map<String, Object> arguments) {
        // 把交换机的名字加上虚拟主机名作为前缀
        exchangeName = getName(exchangeName);
        try {
            synchronized (exchangeLocker) {
                // 1. 判定该交换机是否已经存在, 通过内存查询
                Exchange existsExchange = memoryDataCenter.getExchange(exchangeName);
                if (existsExchange != null) {
                    System.out.println("[VirtualHost] 交换机已经存在! exchangeName=" + exchangeName);
                    return true;
                }
                // 2. 创建交换机
                Exchange exchange = new Exchange();
                exchange.setName(exchangeName);
                exchange.setType(exchangeType);
                exchange.setDurable(durable);
                exchange.setAutoDelete(autoDelete);
                exchange.setArguments(arguments);
                // 3. 把交换机对象写入硬盘
                if (durable) {
                    diskDataCenter.insertExchange(exchange);
                }
                // 4. 把交换机对象写入内存
                memoryDataCenter.insertExchange(exchange);
                System.out.println("[VirtualHost] 交换机创建完成! exchangeName=" + exchangeName);

                // 上述逻辑, 先写硬盘，后写内存。目的就是因为硬盘更容易写失败，如果硬盘失败了，内存就不写了。
                return true;
            }


        } catch (Exception e) {
            e.printStackTrace();
            System.out.println("[VirtualHost] 交换机创建失败! exchangeName=" + exchangeName);
            return false;
        }
    }

    /**
     * 删除交换机
     *
     * @param exchangeName
     * @return
     */
    public boolean exchangeDelete(String exchangeName) {
        exchangeName = getName(exchangeName);
        try {
            synchronized (exchangeLocker) {
                // 1. 先找到对应的交换机
                Exchange toDelete = memoryDataCenter.getExchange(exchangeName);
                if (toDelete == null) {
                    throw new MqException("[VirtualHost] 交换机不存在, 无法删除!");
                }
                // 2. 删除硬盘上的数据
                if (toDelete.isDurable()) {
                    // 是持久化的交换机则才删除
                    diskDataCenter.deleteExchange(exchangeName);
                }
                // 3. 删除内存中的交换机数据
                memoryDataCenter.deleteExchange(exchangeName);
                System.out.println("[VirtualHost] 交换机删除成功! exchangeName=" + exchangeName);
                return true;
            }
        } catch (Exception e) {
            e.printStackTrace();
            System.out.println("[VirtualHost] 交换机删除失败! exchangeName=" + exchangeName);
            return false;
        }
    }


    /**
     * 创建队列
     *
     * @param queueName
     * @param durable
     * @param exclusive
     * @param autoDelete
     * @param arguments
     * @return
     */
    public boolean queueDeclare(String queueName, boolean durable, boolean exclusive, boolean autoDelete,
                                Map<String, Object> arguments) {
        // 把队列的名字给拼接上虚拟主机的名字
        queueName = getName(queueName);
        try {
            synchronized (queueLocker) {
                // 1. 判定队列是否存在
                MsgQueue existsQueue = memoryDataCenter.getQueue(queueName);
                if (existsQueue != null) {
                    System.out.println("[VirtualHost] 队列已经存在! queueName=" + queueName);
                    return true;
                }

                // 2. 创建队列对象
                MsgQueue queue = new MsgQueue();
                queue.setName(queueName);
                queue.setDurable(durable);
                queue.setExclusive(exclusive);
                queue.setAutoDelete(autoDelete);
                queue.setArguments(arguments);

                // 3. 写入银盘
                if (durable) {
                    diskDataCenter.insertQueue(queue);
                }

                // 4. 写入内存
                memoryDataCenter.insertQueue(queue);
                System.out.println("[VirtualHost] 队列创建成功! queueName=" + queueName);
                return true;
            }
        } catch (Exception e) {
            e.printStackTrace();
            System.out.println("[VirtualHost] 队列创建失败! queueName=" + queueName);
            return false;
        }
    }


    /**
     * 删除队列
     *
     * @param queueName
     * @return
     */
    public boolean queueDelete(String queueName) {
        queueName = getName(queueName);
        try {
            synchronized (queueLocker) {
                // 1. 根据队列名字查询下当前队列对象
                MsgQueue queue = memoryDataCenter.getQueue(queueName);
                if (queue == null) {
                    throw new MqException("队列不存在, 无法删除！queueName=" + queueName);
                }

                // 2. 删除硬盘上的队列
                if (queue.isDurable()) {
                    diskDataCenter.deleteQueue(queueName);
                }

                // 3. 删除内存上的队列
                memoryDataCenter.deleteQueue(queueName);
                System.out.println("[VirtualHost] 删除队列成功! queueName=" + queueName);
                return true;
            }
        } catch (Exception e) {
            e.printStackTrace();
            System.out.println("[VirtualHost] 队列删除失败! queueName=" + queueName);
            return false;
        }
    }


    /**
     * 绑定队列和交换机
     *
     * @param queueName
     * @param exchangeName
     * @param bindingKey
     * @return
     */
    public boolean queueBind(String queueName, String exchangeName, String bindingKey) {
        queueName = getName(queueName);
        exchangeName = getName(exchangeName);

        try {
            synchronized (exchangeLocker) {
                synchronized (queueLocker) {
                    // 1. 判定当前的绑定是否已经存在
                    Binding existsBinding = memoryDataCenter.getBinding(exchangeName, queueName);
                    if (existsBinding != null) {
                        throw new MqException("[VirtualHost] 绑定已经存在! queueName=" + queueName +
                                ", exchangeName=" + exchangeName);
                    }

                    // 2. 验证 bindingKey 是否合法
                    if (!router.checkBindingKey(bindingKey)) {
                        throw new MqException("[VirtualHost] bindingKey 非法! bindingKey=" + bindingKey);
                    }

                    // 3. 创建 Binding 对象
                    Binding binding = new Binding();
                    binding.setExchangeName(exchangeName);
                    binding.setQueueName(queueName);
                    binding.setBindingKey(bindingKey);

                    // 4. 获取一下对应的交换机和队列, 如果交换机或者队列不存在, 这样的绑定也是无法创建的
                    MsgQueue queue = memoryDataCenter.getQueue(queueName);
                    if (queue == null) {
                        throw new MqException("[VirtualHost] 队列不存在, 无法创建绑定! queueName=" + queueName);
                    }
                    Exchange exchange = memoryDataCenter.getExchange(exchangeName);
                    if (exchange == null) {
                        throw new MqException("[VirtualHost] 交换机不存在, 无法创建绑定! exchangeName=" + exchangeName);
                    }

                    // 5. 先写硬盘, 交换机队列都持久化, 绑定才需要持久化
                    if (queue.isDurable() && exchange.isDurable()) {
                        diskDataCenter.insertBinding(binding);
                    }

                    // 6. 写入内存
                    memoryDataCenter.insertBinding(binding);
                    System.out.println("[VirtualHost] 绑定创建成功! exchangeName=" + exchangeName +
                            ", queueName" + queueName);
                    return true;
                }
            }

        } catch (Exception e) {
            e.printStackTrace();
            System.out.println("[VirtualHost] 绑定创建失败! queueName=" + queueName +
                    ", exchangeName=" + exchangeName);
            return false;
        }
    }

    /**
     * 解除绑定
     *
     * @param queueName
     * @param exchangeName
     * @return
     */
    public boolean queueUnbind(String queueName, String exchangeName) {
        // 如果用户先删除了队列或者交换机，此时就无法删除绑定了！！！如何解决？
        // 解决方案：注释掉 步骤 2 的代码

        queueName = getName(queueName);
        exchangeName = getName(exchangeName);

        try {
            synchronized (exchangeLocker) {
                synchronized (queueLocker) {
                    // 1. 获取绑定判断是否已经存在
                    Binding binding = memoryDataCenter.getBinding(exchangeName, queueName);
                    if (binding == null) {
                        throw new MqException("[VirtualHost] 解除绑定失败, 绑定不存在! exchangeName=" + exchangeName +
                                ", queueName=" + queueName);
                    }

                    // 2. 获取对应的队列和交换机
//            MsgQueue queue = memoryDataCenter.getQueue(queueName);
//            if (queue == null) {
//                throw new MqException("[VirtualHost] 队列不存在, 无法解除绑定! queueName=" + queueName);
//            }
//            Exchange exchange = memoryDataCenter.getExchange(exchangeName);
//            if (exchange == null) {
//                throw new MqException("[VirtualHost] 交换机不存在, 无法解除绑定! exchangeName=" + exchangeName);
//            }

                    // 3. 删除硬盘上的绑定数据, 无论绑定是否持久化, 都删除, 就算队列和交换机不存在 这个删除也没有副作用
//            if (queue.isDurable() && exchange.isDurable()) {
//                diskDataCenter.deleteBinding(binding);
//            }
                    diskDataCenter.deleteBinding(binding);


                    // 4. 删除内存上的绑定
                    memoryDataCenter.deleteBinding(binding);
                    System.out.println("[VirtualHost] 解除绑定成功! queueName=" + queueName + "exchangeName=" + exchangeName);
                    return true;
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
            System.out.println("[VirtualHost] 解除绑定失败! exchangeName=" + exchangeName +
                    ", queueName=" + queueName);
            return false;
        }
    }


    /**
     * 发送消息到交换机/队列中
     * @param exchangeName
     * @param routingKey
     * @param basicProperties
     * @param body
     * @return
     */
    public boolean basicPublish(String exchangeName, String routingKey, BasicProperties basicProperties, byte[] body) {
        try {
            // 1. 转换交换机的名字
            exchangeName = getName(exchangeName);

            // 2. 检查 routingKey 是否合法
            if (!router.checkRoutingKey(routingKey)) {
                throw new MqException("[VirtualHost] routingKey 非法! routingKey=" + routingKey);
            }

            // 3. 查找交换机对象
            Exchange exchange = memoryDataCenter.getExchange(exchangeName);
            if (exchange == null) {
                throw new MqException("[VirtualHost] 交换机不存在! exchangeName=" + exchangeName);
            }

            // 4. 判断交换机的类型
            if (exchange.getType() == ExchangeType.DIRECT) {
                // 按照直接交换机的方式来转发消息
                // 以 routingKey 作为队列的名字, 直接把消息写入指定的队列中.
                // 此时, 可以无视绑定规则，因为 直接交换机 routingKey 就是要转发的队列名
                String queueName = getName(routingKey);
                // 构造一个消息对象
                Message message = Message.createMessageWithId(routingKey, basicProperties, body);
                // 查找该队列名对应的对象
                MsgQueue queue = memoryDataCenter.getQueue(queueName);
                if (queue == null) {
                    throw new MqException("[VirtualHost] 队列不存在! queueName=" + queueName);
                }
                // 队列存在, 判断该队列是否与该交换机进行绑定了
                Binding binding = memoryDataCenter.getBinding(exchangeName, queueName);
                if (binding == null) {
                    // 这个交换机没有绑定这个队列, 消息直接丢弃掉
                    return true;
                }
                // 该交换机绑定了指定的队列, 则将消息发送给这个队列
                sendMessage(queue, message);
            } else {
                // 按照 fanout 和 topic 的方式来转发
                // 找到该交换机关联的所有绑定, 并遍历所有的绑定对象
                ConcurrentHashMap<String, Binding> bindingsMap = memoryDataCenter.getBindings(exchangeName);
                for (Map.Entry<String, Binding> entry : bindingsMap.entrySet()) {
                    // 获取到绑定对象, 判断对应的队列是否存在
                    Binding binding = entry.getValue();
                    MsgQueue queue = memoryDataCenter.getQueue(binding.getQueueName());
                    if (queue == null) {
                        // 此处不抛异常，可能有多个这样的队列
                        // 不能因为一个队列的失败，影响到其他队列的消息传输
                        System.out.println("[VirtualHost] basicPublish 发送消息时, 发现队列不存在! queueName=" + binding.getQueueName());
                        continue;
                    }
                    // 构造消息对象
                    Message message = Message.createMessageWithId(routingKey, basicProperties, body);
                    // 判定这个消息是否能转发给该队列
                    // 如果是 fanout, 所有绑定的队列都要转发
                    // 如果是 topic, 还需要判定下, bindingKey 和 routingKey 是不是匹配的.
                    if (!router.route(exchange.getType(), binding, message)) {
                        continue;
                    }
                    // 转发消息给队列
                    sendMessage(queue, message);
                }
            }
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            System.out.println("[VirtualHost] 消息发送失败");
            return false;
        }
    }

    /**
     * 发送消息
     * @param queue
     * @param message
     */
    public void sendMessage(MsgQueue queue, Message message) throws IOException, MqException, InterruptedException {
        // 这里发送消息本质就是把消息写入到硬盘和内存上
        int deliverMode = message.getDeliverMode();
        // deliverMode 为 1 表示不持久化, 为 2 表示持久化. rabbitmq
        if (deliverMode == 2) {
            diskDataCenter.sendMessage(queue, message);
        }
        // 写入内存
        memoryDataCenter.sendMessage(queue, message);

        // 通知消费者消费消息
        consumerManager.notifyConsume(queue.getName());
    }


    // 订阅消息.
    // 添加一个队列的订阅者，当队列收到消息之后，就要把消息推送给对应的订阅者。
    /**
     * 添加一个消费者订阅某个指定队列的消息
     * @param consumerTag 消费者的身份标识
     * @param queueName
     * @param autoAck 消息被消费完成后，应答的方式。为 ture 表示自动应答。为 false 手动应答。
     * @param consumer 一个回调函数。此处类型设定成函数式接口，这样后续调用 basicConsume 并且传递参数时，就可以写作 lambda 样子
     * @return
     */
    public boolean basicConsume(String consumerTag, String queueName, boolean autoAck, Consumer consumer) {
        // 构造一个 ConsumerEnv 对象, 把这个对应的队列找到, 再把这个 ConsumerEnv 对象添加到该队列中
        queueName = getName(queueName);
        try {
            consumerManager.addConsumer(consumerTag, queueName, autoAck, consumer);
            System.out.println("[VirtualHost] basicConsume 成功！queueName=" + queueName);
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            System.out.println("[VirtualHost] basicConsume 失败！queueName=" + queueName);
            return false;
        }
    }

    /**
     * ACK 应答消费消息
     * @return
     */
    public boolean basicAck(String queueName, String messageId) {
        queueName = getName(queueName);
        try {
            // 1. 获取到消息和队列
            Message message = memoryDataCenter.getMessage(messageId);
            if (message == null) {
                throw new MqException("[VirtualHost] 要确认的消息不存在! messageId=" + messageId);
            }
            MsgQueue queue = memoryDataCenter.getQueue(queueName);
            if (queue == null) {
                throw new MqException("[VirtualHost] 要确认的队列不存在! queueName=" + queueName);
            }
            // 2. 删除硬盘上的数据
            if (message.getDeliverMode() == 2) {
                diskDataCenter.deleteMessage(queue, message);
                // 硬盘上的数据，删除后，需要判断是否需要垃圾回收
                executeGc(queue);
            }
            // 3. 删除消息中心的数据
            memoryDataCenter.removeMessage(messageId);
            // 4. 删除待确认的集合中的数据
            memoryDataCenter.removeMessageWaitAck(queueName, messageId);
            System.out.println("[VirtualHost] basicAck 成功! 消息被成功确认! queueName=" + queueName
                            + ", messageId" + messageId);
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            System.out.println("[VirtualHost] basicAck 失败! 消息确认失败! queueName=" + queueName
                    + ", messageId" + messageId);
            return false;
        }
    }

    /**
     * 消费者下线
     * @param consumerTag
     */
    public void consumerOffline(String consumerTag) {
        memoryDataCenter.consumerOffline(consumerTag);
    }


    /**
     * 执行 gc 操作
     * @param queue
     * @throws IOException
     * @throws MqException
     * @throws ClassNotFoundException
     */
    public void executeGc(MsgQueue queue) throws IOException, MqException, ClassNotFoundException {
        if (checkGc(queue.getName())) {
            gc(queue);
        }
    }

    /**
     * 由虚拟主机提供 gc 方法，因为硬盘上的 gc 需要涉及到修改内存数据的值
     * @param queue
     * @throws IOException
     * @throws MqException
     * @throws ClassNotFoundException
     */
    private void gc(MsgQueue queue) throws IOException, MqException, ClassNotFoundException {
        synchronized (queue) {
            diskDataCenter.gc(queue);
            // gc 成功之后, 读出最新的文件
            List<Message> messages = diskDataCenter.loadAllMessageFromQueue(queue.getName());
            // 更新内存中的 message 的属性, 因为垃圾回收后, 在文件上的 offsetBeg 和 offsetEnd 的值变了
            for (Message item : messages) {
                Message message = memoryDataCenter.getMessage(item.getMessageId());
                message.setOffsetBeg(item.getOffsetBeg());
                message.setOffsetEnd(item.getOffsetEnd());
            }
        }

    }

    // 判断是否需要垃圾回收
    private boolean checkGc(String queueName) {
        return diskDataCenter.checkGC(queueName);
    }
}
